#include "work_thread.h"
#include "event2/event.h"
#include <unistd.h>
#include <thread>
#include <chrono>
#include <iostream>
#include <cstring>
#include "google/protobuf/dynamic_message.h"
#include "event2/bufferevent.h"
#include "socket_manager.h"
#include "com/simple_logger.h"
#include "work_group.h"
#include "transmit/net_bus.h"
#include "app_context.h"

using namespace std;

void static SEventCb(int fd, short what, void *ctx)
{
    WorkThread *work = (WorkThread *)ctx;
    work->Activated(fd);
}

WorkThread::WorkThread(WorkGroup *parent, int id, NetBus *net_bus)
    : parent_(parent),
      id_(id),
      net_bus_(net_bus)
{
}

WorkThread::~WorkThread()
{
    this->Stop();
    if (pipe_ev_)
    {
        event_del(pipe_ev_);
        event_free(pipe_ev_);
        pipe_ev_ = nullptr;
    }
    while (!sock_manager_list_.empty())
    {
        SocketManager *manager = sock_manager_list_.front();
        delete manager;
        sock_manager_list_.pop_front();
    }
}

bool WorkThread::Init()
{
    // 初始化一对管道，只能在linux系统下使用
    int pipefd[2];
    if (pipe(pipefd))
    {
        return false;
    }
    // pipefd[0]读取管道 pipefd[1]发送管道
    this->pipe_endpoint_ = pipefd[1];
    // 创建管道事件
    event_config *ev_conf = event_config_new();
    event_config_set_flag(ev_conf, EVENT_BASE_FLAG_NOLOCK);
    this->ev_base_ = event_base_new_with_config(ev_conf);
    event_config_free(ev_conf);
    if (!ev_base_)
    {
        return false;
    }

    pipe_ev_ = event_new(this->ev_base_, pipefd[0], EV_READ | EV_PERSIST, SEventCb, this);
    event_add(pipe_ev_, 0);
    return true;
}

void WorkThread::Start()
{
    thread_exit_ = false;
    run_thread_ = new thread(&WorkThread::Run, this);
}

void WorkThread::Stop()
{
    thread_exit_ = true;

    if (run_thread_)
    {
        run_thread_->join();
        delete run_thread_;
        run_thread_ = NULL;
    }
}

void WorkThread::Notify(const char *sign)
{
    // 激活
    int re = write(this->pipe_endpoint_, sign, 1);
    if (re <= 0)
    {
        LOG4CPLUS_ERROR(SimpleLogger::Get()->LoggerRef(), "管道激活失败");
    }
}

void WorkThread::Activated(int fd)
{
    char buf[2] = {0};
    int re = read(fd, buf, 1);
    socket_list_mtx_.lock();
    if (strcmp(buf, "c") == 0) // 通知有新的客户端连接
    {
        // new client connect, create SocketManager
        if (socket_list_.empty())
        {
            socket_list_mtx_.unlock();
            return;
        }
        // 读取一条套接字
        int client_sock = socket_list_.front();
        socket_list_.pop_front();
        // 创建socketManager
        SocketManager *manager = new SocketManager(this, ev_base_, client_sock, AppContext::Get()->client_timeout());
        manager->Init();
        sock_manager_list_.push_back(manager);
        stringstream ss;
        ss << "SocketManager:" << client_sock << " 创建完成";
        LOG4CPLUS_DEBUG(SimpleLogger::Get()->LoggerRef(), ss.str());
    }
    else if (strcmp(buf, "e") == 0) // 通知消息总线上有消息需要来读取
    {
        Metadata *meta = net_bus_->ReadFromHead();
        for (auto it = sock_manager_list_.begin(); it != sock_manager_list_.end(); it++)
        {
            (*it)->ReadFromNetBus(meta);
        }
        // 内部所有连接都读取完成，告知总线ttl减一
        net_bus_->ReadFinish();
    }
    socket_list_mtx_.unlock();
}

void WorkThread::AddSocket(int sock)
{
    socket_list_mtx_.lock();
    this->socket_list_.push_back(sock);
    socket_list_mtx_.unlock();
}

void WorkThread::Disconnect(SocketManager *m)
{
    socket_list_mtx_.lock();
    sock_manager_list_.remove(m);
    delete m;
    socket_list_mtx_.unlock();
}

void WorkThread::SendToNetBus(const char *header, int header_size, const char *data, int data_size)
{
    // 发送消息到消息总线
    net_bus_->Add(header, header_size, data, data_size);
}

int WorkThread::connect_num() const
{
    return sock_manager_list_.size();
}

int WorkThread::work_id() const
{
    return this->id_;
}

void WorkThread::Run()
{
    if (!this->ev_base_)
    {
        return;
    }
    // 内部事件循环
    while (!thread_exit_)
    {
        event_base_loop(this->ev_base_, EVLOOP_NONBLOCK);
        this_thread::sleep_for(chrono::milliseconds(1));
    }
    event_base_free(this->ev_base_);
}